package com.lgl.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.lgl.handler.UserClientHandler;
import com.lgl.pojo.RpcRequest;
import com.lgl.serializer.JSONSerializerImpl;
import com.lgl.service.IUserService;
import com.lgl.tools.RpcEncoder;
import com.lgl.zkclient.ZkClientTools;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class RPCConsumer {

    /** 存储服务器信息的子节点**/
    private static List<Map<String, String>> childServerData = new ArrayList<>();
    private static CuratorFramework client = ZkClientTools.getClient();
    private static List<UserClientHandler> handlerList = new ArrayList<>();

    public static void addChildrenNodesWatcher()  {
        String path = "/";
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, false);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("节点变更了" + pathChildrenCacheEvent.getType());
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                if (type == PathChildrenCacheEvent.Type.CHILD_ADDED
                    || type == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                    initClient();
                    IUserService service = (IUserService)RPCConsumer.createProxy(IUserService.class);
                    String result = service.sayHello("服务器变更!");
                    System.out.println("响应结果：" + result);
                }
            }
        };
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void getChildrenNodeData(CuratorFramework curatorFramework) {
        String path = "/";
        try {
            List<String> strings = curatorFramework.getChildren().forPath(path);
            if (strings == null || strings.size() == 0) {
                return;
            }

            childServerData.clear();
            for (int i = 0; i < strings.size(); i++) {
                String childPath = path + strings.get(i);
                byte[] bytes = curatorFramework.getData().forPath(childPath);
                if (bytes.length > 0) {
                    Map<String, String> map = JSON.parseObject(new String(bytes), new TypeReference<Map<String, String>>() {});
                    childServerData.add(map);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //1. 创建一个线程池对象- 处理自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    //3. 初始化客户端， 连接服务器
    public static void initClient() throws Exception {
        getChildrenNodeData(client);
        if (childServerData != null && childServerData.size() > 0) {
            handlerList.clear();
            for (Map<String, String> map : childServerData) {
                UserClientHandler userClientHandler = new UserClientHandler();
                String serverAddress = map.get("ip");
                Integer port = Integer.valueOf(map.get("port"));
                userClientHandler.setServerPort(String.valueOf(port));

                NioEventLoopGroup group = new NioEventLoopGroup();
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .option(ChannelOption.SO_KEEPALIVE, true)
                        //监听channel 并初始化
                        .handler(new ChannelInitializer<SocketChannel>() {
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializerImpl()));
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(userClientHandler);
                            }
                        });
                handlerList.add(userClientHandler);
                bootstrap.connect(serverAddress, port).sync();
                System.out.println("连接服务器，地址：" + serverAddress + "端口:" + port.toString());
            }
        }
    }

    //4. 编写一个方法，使用JDK的动态代理创建对象
    /**
     *
     * @param serviceClass
     * @return
     */
    public static Object createProxy(Class<?> serviceClass) throws Exception {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass},
                new InvocationHandler() {
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.初始化客户端
                        if (handlerList.size() == 0) {
                            initClient();
                        }
                        StringBuffer result = new StringBuffer();
                        for (UserClientHandler clientHandler : handlerList) {
                            //2.给UserClient设置param参数
                            RpcRequest rpcRequest = createRPCRequest(method, args);
                            clientHandler.setRpcRequest(rpcRequest);
                            rpcRequest.setServerPort(clientHandler.getServerPort());
                            rpcRequest.setSendRpcTime(System.currentTimeMillis());

                            //3.使用线程池，开启一个线程处理call()写操作， 并返回结果
                            String resultTemp = (String) executorService.submit(clientHandler).get();
                            result = result.append(resultTemp);

                        }
                        return result.toString();
                    }
                });
    }

    private static RpcRequest createRPCRequest(Method method, Object[] args) {
        RpcRequest rpcRequest = new RpcRequest();
        rpcRequest.setRequestId(UUID.randomUUID().toString());
        String className = method.getDeclaringClass().getName();
        rpcRequest.setClassName(className);
        rpcRequest.setMethodName(method.getName());
        rpcRequest.setParameterTypes(method.getParameterTypes());
        rpcRequest.setParameters(args);

        return rpcRequest;
    }
}
